package com.bestv.bsr.cloud.service.impl.zt;

import cn.hutool.core.collection.CollUtil;
import com.bestv.bsr.cloud.common.BsrThreadFactory;
import com.bestv.bsr.cloud.common.DisplayNotifyService;
import com.bestv.bsr.core.annotations.OptionalComponent;
import com.bestv.bsr.core.constants.CacheConstants;
import com.bestv.bsr.core.constants.ServiceCodeEnum;
import com.bestv.bsr.core.context.RecommendContext;
import com.bestv.bsr.core.entities.vo.RecDataVo;
import com.bestv.bsr.core.exceptions.ServiceException;
import com.bestv.bsr.core.kafka.BSRMsg;
import com.bestv.bsr.core.kafka.MsgType;
import com.bestv.bsr.core.kafka.dto.RecommendRefreshBody;
import com.bestv.bsr.core.service.impl.zt.AbstractZTRecommendEngine;
import com.bestv.bsr.core.service.impl.zt.dto.ZTRequestParam;
import com.bestv.bsr.core.service.impl.zt.dto.ZTResponseDto;
import com.bestv.bsr.core.service.impl.zt.enums.EngineType;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.apm.toolkit.trace.RunnableWrapper;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author taojiacheng
 */
@Slf4j
@RequiredArgsConstructor
@OptionalComponent(value = "ztRecommendEngine", strategy = OptionalComponent.Strategy.CLOUD)
public class CloudZTRecommendEngine extends AbstractZTRecommendEngine {

    protected final RedissonClient redissonClient;
    protected final BsrThreadFactory bsrThreadFactory;
    protected final DisplayNotifyService displayNotifyService;


    @Override
    public RecDataVo recommend(RecommendContext recommendContext) {
        List<String> itemTypeList = recommendContext.getItemTypes();
        Set<String> leftTypes = new HashSet<>(itemTypeList);
        ZTRequestParam ztRequestParam = buildZtRequestParams(recommendContext);
        ZTResponseDto ztResponseDto;
        EngineType engineType = EngineType.ITEM_BASE;
        if (StringUtils.equals("1", recommendContext.getSceneType())) {
            // 直播推荐场景
            engineType = EngineType.CHANNEL_PREFER;
        }
        if (properties.isCacheEnabled()) {
            // 缓存开启
            String cacheKey = formatCacheKey(recommendContext);
            RBucket<ZTResponseDto> bucket = redissonClient.getBucket(cacheKey);
            ztResponseDto = bucket.get();
            if (ztResponseDto == null) {
                ztResponseDto = ztRemoteService.recommend(ztRequestParam, engineType.getCode(),
                        (result, serverContext, methodContext, paramContext) -> cacheAndNotifyDisplayService(cacheKey, result, recommendContext));
            }
        } else {
            // 缓存未开启
            ztResponseDto = ztRemoteService.recommend(ztRequestParam, engineType.getCode());
            cacheAndNotifyDisplayService("", ztResponseDto, recommendContext);
        }
        if (ztResponseDto != null && !CollectionUtils.isEmpty(ztResponseDto.getData())) {
            // 根据itemType 将数据平均分配
            ztResponseDto = handlerZTResponse(leftTypes, ztResponseDto);
        }
        return convertResponse(ztResponseDto);
    }


    /**
     * 生成缓存并通知展现服务
     */
    protected void cacheAndNotifyDisplayService(String cacheKey, Object result, RecommendContext recommendContext) {
        bsrThreadFactory.execute(RunnableWrapper.of(() -> {
                    if (result instanceof ZTResponseDto) {
                        ZTResponseDto resp = (ZTResponseDto) result;
                        if (CollUtil.isEmpty(resp.getData())) {
                            return;
                        }
                        // 通知
                        try {
                            // 通知逻辑 copy from 长征 bsr-service
                            List<String> itemCodes = resp.getData().stream().map(ZTResponseDto.ZTRecommendItem::getItemCode).collect(Collectors.toList());
                            BSRMsg.Header header = new BSRMsg.Header(MsgType.RECOMMEND_REFRESH, MsgType.RECOMMEND_REFRESH.getName() + ":" + cacheKey);
                            RecommendRefreshBody body = RecommendRefreshBody.builder()
                                    .userGroupCode(recommendContext.getUserGroupCode())
                                    .ruleCode(recommendContext.getRuleCode())
                                    .itemCodes(itemCodes)
                                    .aiSource(recommendContext.getAiSource())
                                    .build();
                            log.info("notify display service: {}", body);
                            BSRMsg<RecommendRefreshBody> bsrMsg = new BSRMsg<>(header, body);
                            displayNotifyService.notify(bsrMsg);
                        } catch (Exception e) {
                            log.error("notify error: {}", e.getMessage(), e);
                        }
                        if (properties.isCacheEnabled() && StringUtils.isNotBlank(cacheKey)) {
                            // 缓存
                            RBucket<ZTResponseDto> bucket = redissonClient.getBucket(cacheKey);
                            bucket.set(resp, properties.getExpireSecond(), TimeUnit.SECONDS);
                        }
                    }
                })
        );
    }

    protected String formatCacheKey(RecommendContext recommendContext) {
        String itemCode = CollectionUtils.isEmpty(recommendContext.getItemCodes()) ?
                StringUtils.EMPTY : recommendContext.getItemCodes().get(0);
        return CacheConstants.format(CacheConstants.AH_CACHE_TEMPLATE,
                recommendContext.getAiSource(),
                recommendContext.getMethodType(),
                recommendContext.getSceneId(),
                // todo itemType 怎么取值？？？
                recommendContext.getItemTypes().get(0),
                recommendContext.getUserId(),
                itemCode
        );
    }


    @Override
    public void validate(RecommendContext recommendContext) {
        if (StringUtils.isEmpty(recommendContext.getMac())) {
            log.error("mac is blank");
            throw new ServiceException(ServiceCodeEnum.MAC_IS_BLANK);
        }
        if (StringUtils.isEmpty(recommendContext.getUserAccount())) {
            log.error("userAccount is blank");
            throw new ServiceException(ServiceCodeEnum.USER_ACCOUNT_IS_BLANK);
        }
    }
}
